package com.torfox.services;

import java.lang.ref.Reference;
import java.lang.ref.ReferenceQueue;
import java.lang.ref.WeakReference;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Set;

import javax.jms.Message;
import javax.jms.MessageListener;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import org.apache.activemq.command.ActiveMQObjectMessage;
import org.apache.activemq.command.ProducerId;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.log4j.spi.LocationInfo;
import org.apache.log4j.spi.LoggingEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Lazy;
import org.springframework.orm.jpa.JpaTemplate;
import org.springframework.stereotype.Service;

import com.torfox.entities.LogRecord;
import com.torfox.modules.livelog.LiveLogConnection;
import com.torfox.modules.livelog.LiveLogConnectionEvent;

@Slf4j
@Service
@Lazy(false)
public class LogListenerService implements MessageListener, ApplicationListener<LiveLogConnectionEvent>
{
	@Autowired @Setter
	private JpaTemplate jpaTemplate;

	private final int hashBufferLimit = 100;

	private final LinkedList<Integer> recentMessageHashes = new LinkedList<Integer>();

	private final ReferenceQueue<ILogListener> rq = new ReferenceQueue<ILogListener>();

	private final Set<WeakReference<ILogListener>> listeners = new HashSet<WeakReference<ILogListener>>();

	@Override
	public void onMessage(final Message message)
	{
		try
		{
			// receive log event in your consumer
			final ActiveMQObjectMessage activeMQMessage = (ActiveMQObjectMessage) message;
			final LoggingEvent event = (LoggingEvent) activeMQMessage.getObject();
			final ProducerId producerId = activeMQMessage.getProducerId();
			final int hash = buildHash( event, producerId);
			synchronized(recentMessageHashes)
			{
				if( recentMessageHashes.contains(hash))
				{
					return;
				}
				recentMessageHashes.offer(hash);
				if( recentMessageHashes.size() > hashBufferLimit)
				{
					recentMessageHashes.poll();
				}
			}
			log.debug("Received log [" + event.getLevel() + "]: " + event.getMessage());
			final LogRecord logRecord = new LogRecord(event, producerId);
			jpaTemplate.persist(logRecord);

			publishRecord( logRecord);
		}
		catch (final Exception e)
		{
			log.warn("Error handling log event", e);
		}
	}

	private int buildHash(final LoggingEvent event, final ProducerId producerId)
    {
		final LocationInfo locationInfo = event.getLocationInformation();
		HashCodeBuilder builder = new HashCodeBuilder().append(event.getTimeStamp()).append(event.getLoggerName()).append(producerId.getConnectionId());
	    if( locationInfo != null)
        {
	        builder = builder.append(locationInfo.getClassName()).append(locationInfo.getMethodName()).append(locationInfo.getLineNumber());
        }
	    return builder.toHashCode();
    }

	private void publishRecord(final LogRecord logRecord)
    {
		Reference<? extends ILogListener> ref = null;
	    while((ref = rq.poll()) != null)
        {
	        listeners.remove(ref);
        }
	    for( final WeakReference<ILogListener> listener : listeners)
        {
	        listener.get().onLog(logRecord);
        }
    }

	@Override
    public void onApplicationEvent(final LiveLogConnectionEvent event)
    {
		final LiveLogConnection connection = event.getConnection();
		if( event instanceof LiveLogConnectionEvent)
		{
			listeners.add( new WeakReference<ILogListener>(connection, rq));
		}
    }

}
